#include "message_server.h"

#include <functional>
#include <unordered_set>

#include "base.pb.h"
#include "log.h"
#include "message.hxx"
#include "uuid.h"

namespace chat {
// MsgStorageServiceImpl

void MsgStorageServiceImpl::GetHistoryMsg(
    ::google::protobuf::RpcController* controller,
    const ::chat::GetHistoryMsgReq* request, ::chat::GetHistoryMsgRsp* response,
    ::google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);

    LOG_DEBUG("收到获取历史消息请求: {}", request->request_id());

    // 取出请求关键信息
    const std::string& request_id = request->request_id();
    const std::string& chat_session_id = request->chat_session_id();
    const int64_t start_time = request->start_time();
    const int64_t end_time = request->over_time();

    bool is_success = true;
    std::string err_msg = "OK";

    do {
        // 获取指定时间段的消息
        std::vector<Message> messages = message_table_->range(
            chat_session_id, boost::posix_time::from_time_t(start_time),
            boost::posix_time::from_time_t(end_time));

        // 对于图片/文件/语音消息，从messages中获取文件id
        std::unordered_set<std::string> file_ids;
        for (auto& message : messages) {
            if (message.file_id() && (*message.file_id()).size() != 0) {
                file_ids.insert(*message.file_id());
            }
        }
        LOG_DEBUG("获取到的文件数量: {}", file_ids.size());

        // 从文件服务获取文件信息
        google::protobuf::Map<std::string, chat::FileDownloadData> file_map;
        if (!getFiles(request_id, file_ids, &file_map)) {
            LOG_ERROR("获取文件信息失败");
            is_success = false;
            err_msg = "获取文件信息失败";

            break;
        }

        // 从用messages获取发送者id
        std::unordered_set<std::string> sender_ids;
        for (auto& message : messages) {
            sender_ids.insert(message.user_id());
        }

        // 从用户服务获取用户信息
        google::protobuf::Map<std::string, chat::UserInfo> user_map;
        if (!getSenders(request_id, sender_ids, &user_map)) {
            LOG_ERROR("获取用户信息失败");
            is_success = false;
            err_msg = "获取用户信息失败";

            break;
        }

        // 遍历所有消息，根据消息类型，填充response
        for (auto& message : messages) {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(message.message_id());
            message_info->set_chat_session_id(message.chat_session_id());
            message_info->set_timestamp(
                boost::posix_time::to_time_t(message.created_time()));
            message_info->mutable_sender()->CopyFrom(
                user_map[message.user_id()]);

            MessageContent* content = message_info->mutable_message();

            switch (message.message_type()) {
                case MessageType::STRING: {
                    content->set_message_type(MessageType::STRING);
                    content->mutable_string_message()->set_content(
                        *message.message_content());
                } break;
                case MessageType::IMAGE: {
                    content->set_message_type(MessageType::IMAGE);
                    content->mutable_image_message()->set_file_id(
                        *message.file_id());
                    content->mutable_image_message()->set_image_content(
                        file_map[*message.file_id()].file_content());
                } break;
                case MessageType::FILE: {
                    content->set_message_type(MessageType::FILE);
                    content->mutable_file_message()->set_file_id(
                        *message.file_id());
                    content->mutable_file_message()->set_file_contents(
                        file_map[*message.file_id()].file_content());
                    content->mutable_file_message()->set_file_name(
                        *message.file_name());
                    content->mutable_file_message()->set_file_size(
                        *message.file_size());
                } break;
                case MessageType::SPEECH: {
                    content->set_message_type(MessageType::SPEECH);
                    content->mutable_speech_message()->set_file_id(
                        *message.file_id());
                    content->mutable_speech_message()->set_file_contents(
                        file_map[*message.file_id()].file_content());
                } break;
                default:
                    LOG_ERROR("{} 未知消息类型", message.message_id());
                    break;
            }
        }

    } while (0);

    response->set_request_id(request_id);
    response->set_success(is_success);
    response->set_errmsg(err_msg);

    LOG_DEBUG("成功处理获取历史消息请求: {}", request_id);
}

void MsgStorageServiceImpl::GetRecentMsg(
    ::google::protobuf::RpcController* controller,
    const ::chat::GetRecentMsgReq* request, ::chat::GetRecentMsgRsp* response,
    ::google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);

    LOG_DEBUG("收到获取最近消息请求: {}", request->request_id());

    // 取出请求关键信息
    const std::string& request_id = request->request_id();
    const std::string& chat_session_id = request->chat_session_id();
    const int64_t count = request->msg_count();

    bool is_success = true;
    std::string err_msg = "OK";

    do {
        // 获取指定时间段的消息
        std::vector<Message> messages =
            message_table_->recent(chat_session_id, count);

        // 对于图片/文件/语音消息，从messages中获取文件id
        std::unordered_set<std::string> file_ids;
        for (auto& message : messages) {
            if (message.file_id() && (*message.file_id()).size() != 0) {
                file_ids.insert(*message.file_id());
            }
        }

        // 从文件服务获取文件信息
        google::protobuf::Map<std::string, chat::FileDownloadData> file_map;
        if (!getFiles(request_id, file_ids, &file_map)) {
            LOG_ERROR("获取文件信息失败");
            is_success = false;
            err_msg = "获取文件信息失败";

            break;
        }

        // 从用messages获取发送者id
        std::unordered_set<std::string> sender_ids;
        for (auto& message : messages) {
            sender_ids.insert(message.user_id());
        }

        // 从用户服务获取用户信息
        google::protobuf::Map<std::string, chat::UserInfo> user_map;
        if (!getSenders(request_id, sender_ids, &user_map)) {
            LOG_ERROR("获取用户信息失败");
            is_success = false;
            err_msg = "获取用户信息失败";

            break;
        }

        // 遍历所有消息，根据消息类型，填充response
        for (auto& message : messages) {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(message.message_id());
            message_info->set_chat_session_id(message.chat_session_id());
            message_info->set_timestamp(
                boost::posix_time::to_time_t(message.created_time()));
            message_info->mutable_sender()->CopyFrom(
                user_map[message.user_id()]);

            MessageContent* content = message_info->mutable_message();

            switch (message.message_type()) {
                case MessageType::STRING: {
                    content->set_message_type(MessageType::STRING);
                    content->mutable_string_message()->set_content(
                        *message.message_content());
                } break;
                case MessageType::IMAGE: {
                    content->set_message_type(MessageType::IMAGE);
                    content->mutable_image_message()->set_file_id(
                        *message.file_id());
                    content->mutable_image_message()->set_image_content(
                        file_map[*message.file_id()].file_content());
                } break;
                case MessageType::FILE: {
                    content->set_message_type(MessageType::FILE);
                    content->mutable_file_message()->set_file_id(
                        *message.file_id());
                    content->mutable_file_message()->set_file_contents(
                        file_map[*message.file_id()].file_content());
                    content->mutable_file_message()->set_file_name(
                        *message.file_name());
                    content->mutable_file_message()->set_file_size(
                        *message.file_size());
                } break;
                case MessageType::SPEECH: {
                    content->set_message_type(MessageType::SPEECH);
                    content->mutable_speech_message()->set_file_id(
                        *message.file_id());
                    content->mutable_speech_message()->set_file_contents(
                        file_map[*message.file_id()].file_content());
                } break;
                default:
                    LOG_ERROR("{} 未知消息类型", message.message_id());
                    break;
            }
        }

    } while (0);

    response->set_request_id(request_id);
    response->set_success(is_success);
    response->set_errmsg(err_msg);

    LOG_DEBUG("成功处理获取最近消息请求: {}", request_id);
}

void MsgStorageServiceImpl::MsgSearch(
    ::google::protobuf::RpcController* controller,
    const ::chat::MsgSearchReq* request, ::chat::MsgSearchRsp* response,
    ::google::protobuf::Closure* done) {
    brpc::ClosureGuard closure_guard(done);

    LOG_DEBUG("收到搜索文本消息请求: {}", request->request_id());

    // 取出请求关键信息
    const std::string& request_id = request->request_id();
    const std::string& chat_session_id = request->chat_session_id();
    const std::string& keyword = request->search_key();

    bool is_success = true;
    std::string err_msg = "OK";

    do {
        // 从 ES 中搜索消息 (只能是文本消息)
        std::vector<Message> messages =
            es_message_->search(keyword, chat_session_id);

        // 从用messages获取发送者id
        std::unordered_set<std::string> sender_ids;
        for (auto& message : messages) {
            sender_ids.insert(message.user_id());
        }

        // 从用户服务获取用户信息
        google::protobuf::Map<std::string, chat::UserInfo> user_map;
        if (!getSenders(request_id, sender_ids, &user_map)) {
            LOG_ERROR("获取用户信息失败");
            is_success = false;
            err_msg = "获取用户信息失败";

            break;
        }

        // 遍历所有消息，填充response
        for (auto& message : messages) {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(message.message_id());
            message_info->set_chat_session_id(message.chat_session_id());
            message_info->set_timestamp(
                boost::posix_time::to_time_t(message.created_time()));
            message_info->mutable_sender()->CopyFrom(
                user_map[message.user_id()]);

            MessageContent* content = message_info->mutable_message();
            content->set_message_type(MessageType::STRING);
            content->mutable_string_message()->set_content(
                *message.message_content());
        }

    } while (0);

    response->set_request_id(request_id);
    response->set_success(is_success);
    response->set_errmsg(err_msg);

    LOG_DEBUG("成功处理搜索文本消息请求: {}", request_id);
}

void MsgStorageServiceImpl::onMessage(const char* message, size_t size) {
    LOG_DEBUG("收到消息进行存储");

    // 反序列化，解析消息
    MessageInfo msg;
    if (!msg.ParseFromArray(message, size)) {
        LOG_ERROR("反序列化失败");
        return;
    }

    const std::string& chat_session_id = msg.chat_session_id();
    const std::string& message_id = msg.message_id();
    const boost::posix_time::ptime& timestamp =
        boost::posix_time::from_time_t(msg.timestamp());
    const std::string& sender_id = msg.sender().user_id();
    const MessageType message_type = msg.message().message_type();

    std::string file_id;
    std::string file_content;  // 只存储文本消息

    // 只存储文件消息
    std::string file_name;
    unsigned long file_size = 0;

    // 文本消息，存入 ES 中，用于关键字查询寻
    // 图片/语音/文件，上传到文件服务，并获取文件id
    switch (message_type) {
        case MessageType::STRING: {
            file_content = msg.message().string_message().content();

            if (!es_message_->appendData(chat_session_id, message_id, timestamp,
                                         file_content, sender_id)) {
                LOG_ERROR("ES 写入文本数据失败, message id: {}", message_id);
                return;
            }
        } break;
        case MessageType::IMAGE: {
            const std::string& content =
                msg.message().image_message().image_content();

            if (!putFile("image" + file_id, content.size(), content,
                         &file_id)) {
                LOG_ERROR("{} 图片文件上传失败", file_id);
                return;
            }
        } break;
        case MessageType::FILE: {
            file_name = msg.message().file_message().file_name();
            file_size = msg.message().file_message().file_size();
            const std::string& file_content =
                msg.message().file_message().file_contents();

            if (!putFile(file_name, file_size, file_content, &file_id)) {
                LOG_ERROR("{} 文件上传失败", file_id);
                return;
            }
        } break;
        case MessageType::SPEECH: {
            const std::string& content =
                msg.message().speech_message().file_contents();

            if (!putFile("speech" + file_id, content.size(), content,
                         &file_id)) {
                LOG_ERROR("{} 语音文件上传失败", file_id);
                return;
            }
        } break;
        default:
            LOG_ERROR("未知消息类型");
            return;
    }

    // 将消息的元数据存入myql
    Message msg_info(message_id, chat_session_id, sender_id, timestamp,
                     message_type);
    msg_info.file_id(file_id);
    msg_info.message_content(file_content);
    msg_info.file_name(file_name);
    msg_info.file_size(file_size);

    if (!message_table_->insert(msg_info)) {
        LOG_ERROR("{} 消息元数据插入失败", message_id);
    }

    LOG_DEBUG("{} 消息元数据插入成功", message_id);
}

bool MsgStorageServiceImpl::putFile(const std::string& file_name,
                                    const unsigned long file_size,
                                    const std::string& file_content,
                                    std::string* file_id) {
    ServiceChannel::ChannelPtr channel =
        service_manager_->choose(file_service_name_);
    if (nullptr == channel) {
        LOG_ERROR("未找到能够提供服务 {} 的节点", file_service_name_);

        return false;
    }

    // 4. 发起Rpc调用
    FileService_Stub stub(channel.get());
    PutSingleFileReq req;
    PutSingleFileRsp rsp;

    req.set_request_id(Uuid::uuid());
    auto file_data = req.mutable_file_data();
    file_data->set_file_name(file_name);
    file_data->set_file_content(file_content);
    file_data->set_file_size(file_size);

    brpc::Controller cntl;
    stub.PutSingleFile(&cntl, &req, &rsp, nullptr);
    if (cntl.Failed()) {
        LOG_ERROR("{}: {} 文件上传RPC调用失败", req.request_id(), file_name);

        return false;
    }
    if (!rsp.success()) {
        LOG_ERROR("{}: {} 文件上传响应错误", req.request_id(), file_name);

        return false;
    }

    *file_id = rsp.file_info().file_id();

    return true;
}

bool MsgStorageServiceImpl::getFiles(
    const std::string& request_id,
    const std::unordered_set<std::string>& file_ids,
    google::protobuf::Map<std::string, chat::FileDownloadData>* file_map) {
    chat::ServiceChannel::ChannelPtr channel =
        service_manager_->choose(file_service_name_);
    if (nullptr == channel) {
        LOG_ERROR("{}: 未找到能够提供服务 {} 的节点", request_id,
                  file_service_name_);

        return false;
    }

    // 4. 先发起Rpc调用，进行批量文件下载
    chat::FileService_Stub stub(channel.get());
    chat::GetMultiFileReq req;
    chat::GetMultiFileRsp rsp;
    req.set_request_id(Uuid::uuid());
    for (auto& file_id : file_ids) {
        req.add_file_id_list(file_id.c_str());
    }

    brpc::Controller cntl;
    stub.GetMultiFile(&cntl, &req, &rsp, nullptr);
    if (cntl.Failed()) {
        LOG_ERROR("{}: 文件批量下载RPC调用失败", request_id);

        return false;
    }
    if (!rsp.success()) {
        LOG_ERROR("{}: 文件批量下载响应错误", request_id);

        return false;
    }

    *file_map = rsp.file_data();

    return true;
}

bool MsgStorageServiceImpl::getSenders(
    const std::string& request_id,
    const std::unordered_set<std::string>& user_ids,
    google::protobuf::Map<std::string, chat::UserInfo>* user_map) {
    chat::ServiceChannel::ChannelPtr channel =
        service_manager_->choose(user_service_name_);
    if (nullptr == channel) {
        LOG_ERROR("{}: 未找到能够提供服务 {} 的节点", request_id,
                  user_service_name_);

        return false;
    }

    // 4. 先发起Rpc调用，进行批量文件下载
    chat::UserService_Stub stub(channel.get());
    chat::GetMultiUserInfoReq req;
    chat::GetMultiUserInfoRsp rsp;
    req.set_request_id(Uuid::uuid());
    for (auto& user_id : user_ids) {
        req.add_users_id(user_id.c_str());
    }

    brpc::Controller cntl;
    stub.GetMultiUserInfo(&cntl, &req, &rsp, nullptr);
    if (cntl.Failed()) {
        LOG_ERROR("{}: 用户信息查询RPC调用失败", request_id);

        return false;
    }
    if (!rsp.success()) {
        LOG_ERROR("{}: 用户信息查询响应错误", request_id);

        return false;
    }

    *user_map = rsp.users_info();

    return true;
}

// -------------------------------------------------------------------

// MsgStorageServerBuilder
void MsgStorageServerBuilder::makeRegisterClient(
    const std::string& register_name, const std::string& etcd_host,
    const std::string& access_host) {
    register_client_ = std::make_shared<Registry>(etcd_host);

    register_client_->registry(register_name, access_host);
}

void MsgStorageServerBuilder::makeServiceManager(
    const std::string& file_service_name,
    const std::string& user_service_name) {
    service_manager_ = std::make_shared<ServiceManager>();
    service_manager_->declared(file_service_name);  // 添加关心
    service_manager_->declared(user_service_name);  // 添加关心

    file_service_name_ = file_service_name;
    user_service_name_ = user_service_name;
}

void MsgStorageServerBuilder::makeDiscoveryClient(
    const std::string& etcd_host, const std::string& base_service) {
    discovery_client_ = std::make_shared<Discovery>(
        etcd_host, base_service,
        std::bind(&ServiceManager::onServiceOnline, service_manager_.get(),
                  std::placeholders::_1, std::placeholders::_2),
        std::bind(&ServiceManager::onServiceOffline, service_manager_.get(),
                  std::placeholders::_1, std::placeholders::_2));
}

void MsgStorageServerBuilder::makeBrpcServer(const uint16_t port,
                                             const int idle_timeout_sec,
                                             const int num_threads) {
    // 检查
    if (odb_client_ == nullptr) {
        LOG_ERROR("ODB 客户端未初始化");
        abort();
    }
    if (es_client_ == nullptr) {
        LOG_ERROR("ES 客户端未初始化");
        abort();
    }
    if (mq_client_ == nullptr) {
        LOG_ERROR("MQ 客户端未初始化");
        abort();
    }
    if (service_manager_ == nullptr) {
        LOG_ERROR("服务管理对象 未初始化");
        abort();
    }

    brpc_server_ = std::make_shared<brpc::Server>();
    MsgStorageServiceImpl* storage_service = new MsgStorageServiceImpl(
        user_service_name_, file_service_name_, service_manager_, mq_client_,
        exchange_name_, routing_key_, odb_client_, es_client_);
    if (-1 ==
        brpc_server_->AddService(storage_service,
                                 brpc::ServiceOwnership::SERVER_OWNS_SERVICE)) {
        LOG_ERROR("RPC添加服务失败");
        abort();
    }
    mq_client_->consume(
        queue_name_, "message-consume",
        std::bind(&MsgStorageServiceImpl::onMessage, storage_service,
                  std::placeholders::_1, std::placeholders::_2));

    brpc::ServerOptions options;
    options.idle_timeout_sec = idle_timeout_sec;
    options.num_threads = num_threads;

    if (-1 == brpc_server_->Start(port, &options)) {
        LOG_ERROR("RPC服务启动失败");
        abort();
    }
}

void MsgStorageServerBuilder::makeOdbClient(
    const int max_pool, const std::string& user, const std::string& password,
    const std::string& db, const std::string& ip, const uint16_t port,
    const std::string& charset) {
    odb_client_ = DataBaseFactory::create(max_pool, user, password, db, ip,
                                          port, charset);
}

void MsgStorageServerBuilder::makeEsClient(
    const std::vector<std::string> host_list) {
    es_client_ = ESClientFactory::create(host_list);
}

void MsgStorageServerBuilder::makeMQClient(const std::string& user,
                                           const std::string& password,
                                           const std::string& host,
                                           const std::string& exchange_name,
                                           const std::string& queue_name,
                                           const std::string& routing_key) {
    exchange_name_ = exchange_name;
    queue_name_ = queue_name;
    routing_key_ = routing_key;
    mq_client_ = std::make_shared<MQClient>(user, password, host);
    mq_client_->declareCompoents(exchange_name, queue_name,
                                 AMQP::ExchangeType::direct, routing_key);
}

MsgStorageServerPtr MsgStorageServerBuilder::build() {
    if (register_client_ == nullptr) {
        LOG_FATAL("服务注册客户端未实例化");
        abort();
    }
    if (discovery_client_ == nullptr) {
        LOG_FATAL("服务发现客户端未实例化");
        abort();
    }
    if (brpc_server_ == nullptr) {
        LOG_FATAL("RPC服务未实例化");
        abort();
    }

    return std::make_shared<MsgStorageServer>(
        register_client_, discovery_client_, brpc_server_, odb_client_);
}
}  // namespace chat